package com.senta.elasticsearch.utils;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.IService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.IndexQuery;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

/**
 * @author anyang
 * Email: 1300378587@qq.com
 * Created Date:2019-11-21 17:18
 */
@Slf4j
public class EsUtil <T> implements ElasticUtil<T>{

    private Class<T> aClass;
    private String indexName;
    private String docType ;
    private String keyField;

    public void setElasticsearchTemplate(ElasticsearchTemplate elasticsearchTemplate){
        this.elasticsearchTemplate = elasticsearchTemplate;
    }

    private ElasticsearchTemplate elasticsearchTemplate;


    EsUtil(Class<T> aClass,String indexName,String docType,String keyField){
        this.docType = docType;
        this.indexName = indexName;
        this.aClass = aClass;
        this.keyField = keyField;
    }
    EsUtil(Class<T> aClass,String indexName,String docType,String keyField,ElasticsearchTemplate elasticsearchTemplate){
        this.elasticsearchTemplate = elasticsearchTemplate;
        this.docType = docType;
        this.indexName = indexName;
        this.aClass = aClass;
        this.keyField = keyField;
    }

    /**
     * 测试结果为数据适配在10w-15w之间，随着数据深分页的增加，速度减慢，由150000/30s --> 150000/75s
     * 其中90%的时间消耗在MySQL，但是由于Elasticsearch是异步执行数据的，初步期望数据流量在10M/s以内为最佳
     * 另，RocketMQ的消息传递在单条消息有限制，10M以内，所以建议单条数据不宜过大，当然在实时通过过程中一般
     * 不会遇到如此大的数据量，暂不设置限制
     *
     * 该方法仅在初始化导入数据时使用
     * @param cacheCount      缓存的每页数据条数
     * @param elasticsearchTemplate      ElasticsearchTemplate 是springboot集成后自带的建议模板
     * @param service                 输入用户的尸体service实现，依赖MybatisPlus
     * @return
     */
    public boolean doDataToEsFromDB(final long cacheCount, ElasticsearchTemplate elasticsearchTemplate, IService<T> service){
        setElasticsearchTemplate(elasticsearchTemplate);
        //当目标索引库存在进行以下操作
        if(elasticsearchTemplate.indexExists(aClass)){

            if(cacheCount>150000L){
                log.warn("缓存容量偏大，建议单次数据小于20M，建议10w条/次");
            }

            //查询次数
            int currentSelectTime = 1;
            //导入次数
            int currentIndexTime = 1;
            //定义当前页数
            long currenPage = 1;
            //获取对应的类
            final int totalCount = service.count();
            final long pageCount = totalCount/cacheCount +1;

            while(currenPage<=pageCount){
                //开始查询记录 准备
                Page<T> billingItemDisPage = new Page<>();
                billingItemDisPage.setSize(cacheCount);
                billingItemDisPage.setCurrent(currenPage);
                billingItemDisPage.setTotal(totalCount);
                log.info("第"+currentSelectTime+"次开始从数据库查询 "+ LocalDateTime.now());
                IPage<T> page = service.page(billingItemDisPage);
                log.info("第"+currentSelectTime+"次结束从数据库查询 "+ LocalDateTime.now());

                //取出对象中的数据

                List<T> dbList = page.getRecords();
                //保存到es
                if(dbList != null && dbList.size()>0){  //判断查询有结果
                    ArrayList<IndexQuery> queries = new ArrayList<>();
                    dbList.forEach(item -> {
                        Object id = null;
                        try{
                             Class<?> theClass = item.getClass();
                             Method method = theClass.getMethod(keyField, null);
                             Object invoke = method.invoke(item, null);
                            id = invoke;
                        }catch(NoSuchMethodException | InvocationTargetException | IllegalAccessException e){
                            log.error(e.getMessage());
                        }
                        if(id == null){
                            throw new UnsupportedOperationException("主键值不允许为null");
                        }
                        //创建对象
                        IndexQuery indexQuery = new IndexQuery();
                        indexQuery.setId(id.toString());
                        indexQuery.setObject(item);
                        indexQuery.setIndexName(indexName);
                        indexQuery.setType(docType);
                        //添加到集合
                        queries.add(indexQuery);
                    });
                    log.info("开始第 "+currentIndexTime+" 导入");
                    elasticsearchTemplate.bulkIndex(queries);
                    log.info("导入第"+currentIndexTime+" 已派发");
                }
                //重定义索引范围
                currentIndexTime++;
                //重定义下一次搜索次数
                currentSelectTime++;
                //设置下一页
                currenPage = page.getCurrent() + 1L;
            }
            log.info("数据库迁移拷贝完成");
            return true;
        }
        return false;
    }

    public boolean createIndex(){
            return doCreateIndex();
    }

    @Override
    public boolean deleteIndex(){
        return elasticsearchTemplate.deleteIndex(indexName);
    }

    public boolean doCreateIndex() {
        boolean result = false;
        if(elasticsearchTemplate == null){
              log.error("缺elasticsearchTemplate对象 -- null");
        }else{
            if(elasticsearchTemplate.indexExists(indexName)){
                boolean delResult = elasticsearchTemplate.deleteIndex(indexName);
                if(delResult){
                 result = elasticsearchTemplate.createIndex(aClass);
                }
            }else
              result = elasticsearchTemplate.createIndex(aClass);
        }
          return result;
    }
}
